
package com.trydofor.id.lazy;

import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.annotation.Nonnull;
import javax.annotation.concurrent.ThreadSafe;

import com.trydofor.id.a9cq.ExtedEntryA9cq;
import com.trydofor.id.core.A9CQ;
import com.trydofor.id.core.A9Loader;
import com.trydofor.id.core.ApiExceptions;

/**
 * an atomic lazy entry a9cq.
 * 
 * <pre>
 * the loader does some actions while calling next(...), if latch is, 
 *  - negative/zero - sync load values if the remaining is empty.
 *  - positive - async load values if the remaining is less than the latch.
 *  
 * if the exception occurs during loading, something happens if the latch is,
 *  - negative/zero - throw a FakeException of ExecutionException to the caller.
 *  - positive - temporarily set the latch to '-1' and call the loader's handleException(e).
 *  
 * and if the next loading is successfully done,
 * the temporarily latch will reset to the original.
 * </pre>
 * 
 * @param <E> the Element of sequence
 * @author : Shi Rongjiu (www.trydofor.com)
 */
@ThreadSafe
public class LazyEntryA9cq<E> implements A9CQ<E> {
    
    private final A9Loader<List<E>> loader;
    private final ThreadFactory     factory;
    private final int               latch;
    
    private final ExtedEntryA9cq<E> entryA9cq = new ExtedEntryA9cq<E>(false, 0);
    private volatile int            procLatch;
    private volatile Future<E>      refFuture;
    
    /**
     * an atomic lazy entry a9cq with default ThreadFactory.
     * 
     * @param loader @Nonnull the loader.
     * @param latch the latch.
     * @see #LazyEntryA9cq(A9Loader, int, ThreadFactory)
     * @see Executors#defaultThreadFactory()
     */
    public LazyEntryA9cq(@Nonnull A9Loader<List<E>> loader, int latch) {
        this(loader, latch, Executors.defaultThreadFactory());
    }
    
    /**
     * an atomic lazy entry a9cq.
     * 
     * @param loader @Nonnull the loader.
     * @param latch the latch.
     * @param factory @Nonnull the thread factory
     * @throws NullPointerException if loader or factory is null
     */
    public LazyEntryA9cq(@Nonnull A9Loader<List<E>> loader, int latch, @Nonnull ThreadFactory factory) {
        ApiExceptions.checkNullPointer(loader);
        ApiExceptions.checkNullPointer(factory);
        
        this.loader = loader;
        this.factory = factory;
        this.latch = latch < 0 ? 0 : latch;
        this.procLatch = this.latch;
    }
    
    /**
     * {@inheritDoc}
     */
    public boolean isAtomic() {
        return true;
    }
    
    /**
     * {@inheritDoc}
     */
    public long remains() {
        return entryA9cq.remains();
    }
    
    /**
     * {@inheritDoc}
     */
    public void reset() {
        entryA9cq.reset();
    }
    
    /**
     * {@inheritDoc}
     */
    public E current() {
        return entryA9cq.current();
    }
    
    /**
     * {@inheritDoc}
     */
    public E next() {
        E r = null;
        try {
            r = next(0, null);
        }
        catch (InterruptedException e) {
            ApiExceptions.throwFakeException(e);
        }
        catch (TimeoutException e) {
            ApiExceptions.throwFakeException(e);
        }
        return r;
    }
    
    /**
     * {@inheritDoc}
     */
    public E next(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
        
        final long startedNano = System.nanoTime();
        final Thread curThread = Thread.currentThread();
        final boolean needTime = (timeout > 0 && unit != null);
        
        E result = entryA9cq.next(timeout, unit);
        while (result == null) { // sync load
            try {
                // blocking
                if (needTime) {
                    asyncLoad().get(timeout, unit);
                }
                else {
                    asyncLoad().get();
                }
            }
            catch (ExecutionException e) {
                ApiExceptions.throwFakeException(e);
            }
            
            if (needTime) {
                ApiExceptions.checkThreadInterrupted(curThread);
                ApiExceptions.checkTimeoutNano(startedNano, timeout, unit);
            }
            
            result = entryA9cq.next(timeout, unit);
        }
        
        // check the cache
        final long remain = entryA9cq.remains();
        if (remain < procLatch) {
            asyncLoad();
        }
        
        return result;
    }
    
    // -----------------
    private Future<E> asyncLoad() {
        //check loading
        final Future<E> current = refFuture;
        if (current != null) return current;
        
        // async load
        ExecutorService executor = Executors.newSingleThreadExecutor(factory);
        final AtomicBoolean await = new AtomicBoolean(true);
        final Future<E> result = executor.submit(new Callable<E>() {
            
            public E call() throws Exception {
                final E nul = null;
                try {
                    List<E> ls = loader.heavyLoad();
                    if (ls != null && ls.size() > 0) {
                        entryA9cq.append(ls);
                    }
                    procLatch = latch;
                }
                catch (Exception e) {
                    if (procLatch > 0) {
                        procLatch = -1;
                        loader.handleException(e);
                    }
                    else {
                        throw e;
                    }
                }
                finally {
                    while (await.get()) {
                        // waiting for 'refFuture = result'
                    }
                    refFuture = null;
                }
                return nul; // only a blocking mark
            }
        });
        
        refFuture = result;
        await.set(false);
        executor.shutdown();
        
        return result;
    }
}
